package storm.starter.comp90056;

import java.io.FileNotFoundException;
import java.io.PrintWriter;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

public class WordCountFrequentAlgBolt extends BaseBasicBolt {

    private static final int MAP_SIZE = 100;

    private final Map<String, Integer> wordCount = new HashMap<String, Integer>(MAP_SIZE);

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        final String word = input.getString(0);
        if (wordCount.containsKey(word)) {
            wordCount.put(word, wordCount.get(word) + 1);
        } else if (wordCount.size() < MAP_SIZE) {
            wordCount.put(word, 1);
        } else {
            final Iterator<Entry<String, Integer>> it = wordCount.entrySet().iterator();
            while (it.hasNext()) {
                final Entry<String, Integer> entry = it.next();
                if (entry.getValue() == 1) {
                    it.remove();
                } else {
                    entry.setValue(entry.getValue() - 1);
                }
            }

        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }

    @Override
    public void cleanup() {
        try {
            final PrintWriter writer = new PrintWriter("output.txt");
            for (final Entry<String, Integer> entry : wordCount.entrySet()) {
                writer.println(entry.getKey() + " " + entry.getValue());
            }
            writer.close();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }

    }

}
